// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: daniel

#pragma once

#include <map>
#include <memory>
#include <optional>
#include <set>
#include <unordered_map>
#include <utility>

#include <pybind11/pybind11.h>

// message build
#include "pipeline/coin_request.pb.h"
#include "pipeline/coin_update.pb.h"

// update
#include "appcoin2/support/py_module/driver/py_order_update.h"
#include "coin2/exchange/base/feed/replay/proto_event.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/order/order.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/strategy/strategy.h"

namespace coin2::exchange::driver {

namespace impl {

namespace py = pybind11;

using coin::proto::OrderContextProto;
using coin::proto::OrderEvent;
using coin::proto::OrderGatewayLog;
using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::feed::FeedUpdateManager;
using coin2::exchange::base::feed::proto::deserialize::BookUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::FundingRateUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::IndexUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::IvGreeksUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::KlineUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::LiquidationUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::MarketAnalysisUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::MarkPriceUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::NavUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::OpenInterestUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::TradeUpdateBuilder;
using coin2::exchange::base::feed::proto::deserialize::WalletUpdateBuilder;
using coin2::exchange::base::feed::timer::FeedTimer;
using coin2::exchange::base::market::MarketExchangeApi;
using coin2::exchange::base::order::OrderContext;
using coin2::exchange::base::order::OrderSpec;
using coin2::exchange::base::order::OrderUpdate;
using coin2::exchange::base::order::OrderUpdateManager;
using coin2::exchange::base::symbology::IProduct;
using exchange::base::order::IOrderSubscriber;
using pipeline::AccountInfoProto;
using pipeline::CoinRequestProto;
using pipeline::CoinUpdateProto;
using pipeline::FeedUpdateProto;
using pipeline::OrderContextBundle;
using pipeline::OrderEventProto;
using pipeline::OrderSnapshotProto;
using pipeline::OrderUpdateProto;
using pipeline::RiskUpdateProto;
using UpdateManagerKey = std::pair<std::string, MarketExchangeApi>;

class RiskUpdateManager {
 public:
  explicit RiskUpdateManager(const std::string& system_name, const MarketExchangeApi& mea)
      : system_name_{system_name}, mea_{mea} {}

  void HandleRiskUpdate(const RiskUpdateProto& upd) {
    switch (upd.type()) {
      case RiskUpdateProto::RISK_EVENT:
        HandleRiskEvent(upd);
        break;
      default:
        break;
    }
  }

  void RegisterSubsciber(IOrderSubscriber* subscriber) {
    // TODO: temporary solution
    subscriber_ = subscriber;
  }

 private:
  void HandleRiskEvent(const RiskUpdateProto& upd) {
    if (!subscriber_) {
      return;
    }
    if (upd.risk_event().risk_event_type() != RiskEvent::RISK_UNDETECTED_POSITION) {
      return;  // skip except risk undetected position first
    }
    OrderGatewayLog og_log;
    og_log.set_type(OrderGatewayLog::RISK_EVENT);
    *og_log.mutable_account_request() = upd.account_request();
    *og_log.mutable_risk_event() = upd.risk_event();
    subscriber_->onOrderLog(og_log);
  }

 private:
  const std::string system_name_;
  const MarketExchangeApi mea_;

  IOrderSubscriber* subscriber_ = nullptr;
};

// TODO(daniel): implement with coin4
class OrderContextManager2 {
 public:
  explicit OrderContextManager2(const MarketExchangeApi& mea, int64_t queue_size = 4096 * 16)
      : mea_{mea} {}

  auto GetAllWorkingOrders() const -> const std::set<const OrderContext*>& {
    return order_context_set_;
  }

  void AddOrderContext(const OrderEvent& order_event) {
    if (OrderEvent::ORDER_SUBMITTED != order_event.type()) {
      return;
    }
    auto proc_order_id = order_event.proc_order_id();
    if (order_context_map_.count(proc_order_id)) {
      return;
    }
    auto product = CreateProductFromNativeString(
        mea_,
        order_event.native_symbol(),
        order_event.order_created_time());

    OrderSpec order_spec(
        *product,
        order_event.order_type(),
        order_event.order_side(),
        order_event.duration(),
        order_event.order_price(),
        order_event.order_qty(),
        order_event.order_created_time(),
        order_event.post_only(),
        0,
        order_event.tag());
    order_spec.proc_order_id = proc_order_id;
    auto order_context = std::make_unique<OrderContext>(order_spec.proc_order_id, order_spec);
    order_context_set_.emplace(order_context.get());
    order_context_map_.emplace(order_context->proc_order_id(), std::move(order_context));
  }

  void DelOrderContext(const OrderEvent& order_event) {
    if (order_event.has_external_order_id()) {
      order_id_to_context_map_.erase(order_event.external_order_id());
    }
    const auto proc_order_id = order_event.proc_order_id();
    if (order_context_map_.count(proc_order_id)) {
      auto& order_context = order_context_map_.at(proc_order_id);
      order_context_set_.erase(order_context.get());
      order_context_map_.erase(proc_order_id);
    }
  }

  void HandleOrderSubmitted(const OrderEvent& order_event) { AddOrderContext(order_event); }

  void HandleOrderAccepted(const OrderEvent& order_event) {
    auto proc_order_id = order_event.proc_order_id();
    if (order_context_map_.count(proc_order_id)) {
      return;
    }
    auto order_id = order_event.external_order_id();
    order_id_to_context_map_[order_id] = order_context_map_.at(proc_order_id).get();
  }

  void HandleOrderRejected(const OrderEvent& order_event) { DelOrderContext(order_event); }

  void HandleOrderFilled(const OrderEvent& order_event) {}

  void HandleCancelConfirmed(const OrderEvent& order_event) { DelOrderContext(order_event); }

  void onOrderEvent(const OrderEvent& order_event) {
    switch (order_event.type()) {
      case OrderEvent::ORDER_SUBMITTED:
        HandleOrderSubmitted(order_event);
        break;
      case OrderEvent::ORDER_ACCEPTED:
        HandleOrderAccepted(order_event);
        break;
      case OrderEvent::ORDER_REJECTED:
        HandleOrderRejected(order_event);
        break;
      case OrderEvent::CANCEL_CONFIRMED:
        HandleCancelConfirmed(order_event);
        break;
      case OrderEvent::ORDER_AUTO_CANCELED:
        HandleCancelConfirmed(order_event);
        break;
      case OrderEvent::ORDER_FILLED:
        HandleOrderFilled(order_event);
        break;
      default:
        break;
    }
  }

  void onOrderConextUpdate(const OrderContextBundle& order_context_bundle) {
    // reset
    order_context_map_.clear();
    order_context_set_.clear();
    order_id_to_context_map_.clear();
    // update
    for (const auto& order_context_proto : order_context_bundle.each_order_context()) {
      auto order_context = OrderContext::FromProto(order_context_proto);
      if (order_context->has_exchange_order_id()) {
        order_id_to_context_map_.emplace(order_context->exchange_order_id(), order_context.get());
      }
      order_context_set_.emplace(order_context.get());
      order_context_map_.emplace(order_context->proc_order_id(), std::move(order_context));
    }
  }

 private:
  const MarketExchangeApi mea_;

  std::unordered_map<int64_t, std::unique_ptr<OrderContext>> order_context_map_;
  std::set<const OrderContext*> order_context_set_;
  std::unordered_map<std::string, const OrderContext*> order_id_to_context_map_;
};

class PyStratForwarder : public coin2::strategy::IStrategy {
 public:
  PyStratForwarder(py::object strat, py::object pipe_incoming, py::object pipe_outgoing)
      : strat_{strat}, pipe_incoming_{pipe_incoming}, pipe_outgoing_{pipe_outgoing} {
    CHECK(py::hasattr(pipe_incoming_, "recv_bytes"));
    CHECK(py::hasattr(pipe_outgoing_, "send_bytes"));
  }

  void Run() {
    while (true) {  // loop poll
      if (!running_) {
        break;
      }
      try {
        while (DoRecv(&recv_buf_)) {
          HandleIncomingMessage(recv_buf_);
          recv_buf_.clear();
        }
      } catch (std::exception& ex) {
        throw py::type_error(fmt::format("cpp exception: {}", ex.what()));
      }
    }
  }

  void Stop() { running_ = false; }

  auto GetAllWorkingOrders(const std::string& system_name) -> const std::vector<PyOrderContext>& {
    // TODO(daniel): add indicator that indicate there is order update
    if (!is_order_updated_) {
      return working_order_;
    }
    working_order_.clear();
    try {
      for (const auto& [os, ocm] : order_context_manager_map_) {
        if (system_name != "ALL" and system_name != os) {
          continue;
        }
        for (const auto* order : ocm->GetAllWorkingOrders()) {
          working_order_.push_back(PyOrderContext(order));
        }
      }
    } catch (std::exception& ex) {
      throw py::type_error(fmt::format("cpp exception: {}", ex.what()));
    }
    return working_order_;
  }

  bool SubmitExecution(py::dict exec_spec) {
    try {
      return HandleExecution(exec_spec);
    } catch (std::exception& ex) {
      throw py::type_error(fmt::format("cpp exception: {}", ex.what()));
      return false;
    }
  }

 private:
  template <class T>
  bool DoSend(T&& msg) {
    pipe_outgoing_.attr("send_bytes")(py::bytes(msg));
    return true;
  }

  template <class T>
  bool DoRecv(T* msg) {
    auto ready = pipe_incoming_.attr("poll")();
    if (py::str(ready).is(py::str(Py_False))) {
      onPollFail();
      return false;
    }
    py::bytes buf = pipe_incoming_.attr("recv_bytes")();
    *msg = std::move(buf);
    return true;
  }

  /**************************************************************************************/
  /* handle outgoing message                                                           */
  /**************************************************************************************/
 private:
  bool HandleExecution(py::dict exec_spec) {
    const auto exec_type_str = exec_spec["exec_type"].cast<std::string>();
    CoinRequestProto::CoinRequestType exec_type;
    CoinRequestProto::CoinRequestType_Parse(exec_type_str, &exec_type);
    switch (exec_type) {
      case CoinRequestProto::SUBMIT_ORDER: {
        auto order_spec = ToOrderSpec(exec_spec["order_spec"]);
        if (!order_spec.has_value()) {
          return false;
        }
        req_proto_.Clear();
        req_proto_.set_type(CoinRequestProto::SUBMIT_ORDER);
        auto* account_request = req_proto_.mutable_account_request();
        account_request->set_system_name(exec_spec["og_name"].cast<std::string>());
        *req_proto_.mutable_submit_order_spec() = (*order_spec).ToProto();
        send_buf_.clear();
        if (!req_proto_.SerializeToString(&send_buf_)) {
          return false;
        }
        return DoSend(send_buf_);
      }
      case CoinRequestProto::CANCEL_ORDER: {
        req_proto_.Clear();
        req_proto_.set_type(CoinRequestProto::CANCEL_ORDER);
        auto* account_request = req_proto_.mutable_account_request();
        account_request->set_system_name(exec_spec["og_name"].cast<std::string>());
        auto* order_spec = req_proto_.mutable_cancel_order_spec();
        order_spec->set_proc_order_id(exec_spec["proc_order_id"].cast<int64_t>());
        send_buf_.clear();
        if (!req_proto_.SerializeToString(&send_buf_)) {
          return false;
        }
        return DoSend(send_buf_);
      }
      case CoinRequestProto::AMEND_ORDER:
        py::print("@daniel: implement amend order");
        return false;
      default:
        py::print(fmt::format("unsupported type: {}", exec_type_str));
        return false;
    }
  }

  auto ToOrderSpec(py::dict exec_spec) const -> std::optional<OrderSpec> {
    using namespace coin2::exchange::base::order;
    using namespace coin::proto;
    try {
      coin::proto::ExchangeType exchange_type;
      const auto exchange_type_str = exec_spec["exchange_type"].cast<std::string>();
      coin::proto::ExchangeType_Parse(exchange_type_str, &exchange_type);

      const auto market_type_str = exec_spec["market_type"].cast<std::string>();
      coin::proto::MarketType market_type;
      coin::proto::MarketType_Parse(market_type_str, &market_type);

      const auto order_type_str = exec_spec["order_type"].cast<std::string>();
      coin::proto::OrderType order_type;
      coin::proto::OrderType_Parse(order_type_str, &order_type);

      const auto order_side_str = exec_spec["order_side"].cast<std::string>();
      coin::proto::OrderSide order_side;
      coin::proto::OrderSide_Parse(order_side_str, &order_side);

      const auto order_duration_str = exec_spec["order_duration"].cast<std::string>();
      coin::proto::OrderDuration order_duration;
      coin::proto::OrderDuration_Parse(order_duration_str, &order_duration);

      const auto norm_product = exec_spec["norm_product"].cast<std::string>();
      const auto timestamp = exec_spec["timestamp"].cast<int64_t>();

      // TODO(daniel): handle the precision of qty
      const auto order_price = exec_spec["order_price"].cast<double>();
      const auto order_qty = exec_spec["order_qty"].cast<double>();

      return OrderSpec(
          exec_spec["product"].cast<std::string>(),
          norm_product,
          exchange_type,
          market_type,
          order_type,
          order_side,
          order_duration,
          order_price,
          order_qty,
          timestamp,
          exec_spec["post_only"].cast<bool>(),
          exec_spec["reduce_only"].cast<bool>(),
          exec_spec["leverage_rate"].cast<double>(),
          exec_spec["tag"].cast<std::string>());
    } catch (std::exception& ex) {
      py::print(fmt::format("failed to convert order spec: {}", ex.what()));
      py::print(exec_spec);
      return {};
    }
  }

  /**************************************************************************************/
  /* handle incoming message                                                           */
  /**************************************************************************************/
 private:
  void HandleIncomingMessage(const std::string& msg) {
    if (!UnserializeMessage(msg)) {
      return;
    }
    switch (upd_proto_.type()) {
      case CoinUpdateProto::FEED_UPDATE:
        HandleFeedUpdate(upd_proto_.feed_update());
        break;
      case CoinUpdateProto::ORDER_UPDATE:
        HandleOrderUpdate(upd_proto_.order_update());
        break;
      case CoinUpdateProto::ACCOUNT_INFO_UPDATE:
        HandleAccountInfo(upd_proto_.account_info_update());
        break;
      case CoinUpdateProto::RISK_UPDATE:
        HandleRiskUpdate(upd_proto_.risk_update());
        break;
      case CoinUpdateProto::SYSTEM_UPDATE:
        py::print(fmt::format("@daniel: handle system update:", upd_proto_.DebugString()));
        break;
      default:
        py::print(fmt::format("unsupported message:\n{}", upd_proto_.DebugString()));
        break;
    }
  }

  bool UnserializeMessage(const std::string& msg) {
    if (msg.empty()) {
      return false;
    }
    try {
      upd_proto_.Clear();
      upd_proto_.ParseFromString(msg);
      return true;
    } catch (std::exception& ex) {
      py::print(fmt::format("invalid message. {}", ex.what()));
      return false;
    }
  }

  auto GetOrCreateProduct(
      const MarketExchangeApi& mea,
      const std::string& symbol,
      int64_t timestamp) -> const IProduct* {
    const auto key = std::pair<std::string, MarketExchangeApi>(symbol, mea);
    auto iter = symbol_to_product_map_.find(key);
    if (iter != symbol_to_product_map_.end()) {
      return iter->second;
    }
    auto product = CreateProductFromNormString(mea, symbol, timestamp, true);
    symbol_to_product_map_.emplace(key, product.get());
    local_product_holder_.emplace_back(std::move(product));
    return local_product_holder_.back().get();
  }

  auto GetOrCreateFeedUpdateManager(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const std::string& symbol,
      int64_t timestamp) -> FeedUpdateManager* {
    const auto key = std::pair<std::string, MarketExchangeApi>(system_name, mea);
    auto iter = feed_upd_mgr_map_.find(key);
    if (iter != feed_upd_mgr_map_.end()) {
      return iter->second;
    }
    auto mgr = std::make_unique<FeedUpdateManager>(system_name, mea, false, false);
    auto* product = GetOrCreateProduct(mea, symbol, timestamp);
    mgr->AddProductForFeed(*product, this);
    feed_upd_mgr_map_.emplace(key, mgr.get());
    feed_upd_mgr_holder_.emplace_back(std::move(mgr));
    return feed_upd_mgr_holder_.back().get();
  }

  void HandleFeedUpdate(const FeedUpdateProto& upd) {
    // py::print(fmt::format("DEBUG: feed: \n{}", upd.DebugString()));
    const auto& account_request = upd.account_request();
    const auto& system_name = account_request.system_name();
    const auto mea = MarketExchangeApi::FromString(fmt::format(
        "{}.{}.{}",
        account_request.market_type(),
        account_request.exchange(),
        account_request.api_version()));
    const auto& proto = upd.fast_feed();
    auto* upd_mgr = GetOrCreateFeedUpdateManager(
        system_name,
        mea,
        proto.symbol() /* absolute_norm */,
        proto.timestamp());
    upd_mgr->mutable_update()->ResetFeedUpdate();
    upd_mgr->mutable_update()->SetTimestamp(proto.timestamp());
    if (!upd_mgr->is_symbol_registered(proto.native_symbol())) {
      auto* product = GetOrCreateProduct(mea, proto.symbol(), proto.timestamp());
      upd_mgr->AddProductForFeed(*product, this);
    }
    switch (proto.feed_type()) {
      case fastfeed::proto::FeedType::FEED_TYPE_BOOK: {
        BookUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishBook();
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_TRADE: {
        TradeUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishTrade(proto.native_symbol(), false);
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_LIQUIDATION_ORDER: {
        LiquidationUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishLiquidationOrder(proto.native_symbol(), false);
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_OPEN_INTEREST: {
        OpenInterestUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishOpenInterest(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_INDEX: {
        IndexUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishIndex(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_MARK_PRICE: {
        MarkPriceUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishMarkPrice(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_FUNDING_RATE: {
        FundingRateUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishFundingRate(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_IV_GREEKS: {
        IvGreeksUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishIvGreeks(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_NAV: {
        NavUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishNav(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_KLINE: {
        // TODO(daniel):
        KlineUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishKline(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_WALLET: {
        WalletUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishWallet(proto.native_symbol());
        break;
      }
      case fastfeed::proto::FeedType::FEED_TYPE_MARKET: {
        MarketAnalysisUpdateBuilder::FromProto(proto, upd_mgr);
        upd_mgr->PublishMarketAnalysis(proto.native_symbol());
        break;
      }
      default:
        py::print(fmt::format("unsupported update: ", upd.DebugString()));
        break;
    }
  }

  auto GetOrCreateOrderUpdateManager(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const std::string& symbol,
      int64_t timestamp) -> OrderUpdateManager* {
    const auto key = std::pair<std::string, MarketExchangeApi>(system_name, mea);
    auto iter = order_upd_mgr_map_.find(key);
    if (iter != order_upd_mgr_map_.end()) {
      return iter->second;
    }
    if ("fatboy" == symbol) {  // for account info update
      auto mgr = std::make_unique<OrderUpdateManager>(system_name, mea);
      mgr->AddNativeSymbol(symbol, this);
      order_upd_mgr_map_.emplace(key, mgr.get());
      order_upd_mgr_holder_.emplace_back(std::move(mgr));
      return order_upd_mgr_holder_.back().get();
    }
    auto mgr = std::make_unique<OrderUpdateManager>(system_name, mea);
    auto* product = GetOrCreateProduct(mea, symbol, timestamp);
    mgr->AddProduct(product);
    mgr->AddNativeSymbol(product->order_native_symbol(), this);
    order_upd_mgr_map_.emplace(key, mgr.get());
    order_upd_mgr_holder_.emplace_back(std::move(mgr));
    return order_upd_mgr_holder_.back().get();
  }

  auto GetOrCreateOrderContextManager(const std::string& system_name, const MarketExchangeApi& mea)
      -> OrderContextManager2* {
    auto iter = order_context_manager_map_.find(system_name);
    if (iter != order_context_manager_map_.end()) {
      return iter->second;
    }
    auto ocm = std::make_unique<OrderContextManager2>(mea, 1024);
    order_context_manager_map_.emplace(system_name, ocm.get());
    order_context_manager_holder_.push_back(std::move(ocm));
    return order_context_manager_holder_.back().get();
  }

  void HandleAccountInfo(const AccountInfoProto& upd) {
    const auto& account_request = upd.account_request();
    const auto& system_name = account_request.system_name();
    const auto mea = MarketExchangeApi::FromString(fmt::format(
        "{}.{}.{}",
        account_request.market_type(),
        account_request.exchange(),
        account_request.api_version()));
    auto timestamp = GetCurrentTimestamp();
    auto* upd_mgr = GetOrCreateOrderUpdateManager(system_name, mea, "fatboy", timestamp);
    switch (upd.type()) {
      case AccountInfoProto::BALANCE: {
        using coin2::exchange::base::executor::TopicRecordPacket;
        TopicRecordPacket packet;
        packet.timestamp = timestamp;
        upd_mgr->onBeginPacket(packet);
        upd_mgr->UpdateAccountBalance(upd.balance(), GetCurrentTimestamp(), true);
        upd_mgr->onEndPacket(packet);
        break;
      }
      case AccountInfoProto::POSITION: {
        using coin2::exchange::base::executor::TopicRecordPacket;
        TopicRecordPacket packet;
        packet.timestamp = timestamp;
        upd_mgr->onBeginPacket(packet);
        upd_mgr->UpdateAccountPosition(upd.position(), GetCurrentTimestamp(), true);
        upd_mgr->onEndPacket(packet);
        break;
      }
      default:
        break;
    }
  }

  void HandleOrderEvent(const OrderEventProto& order_event_proto) {
    const auto& order_event = order_event_proto.order_event();
    switch (order_event.type()) {
      case OrderEvent::ORDER_SUBMITTED:
      case OrderEvent::ORDER_ACCEPTED:
      case OrderEvent::ORDER_REJECTED:
      case OrderEvent::CANCEL_SUBMITTED:
      case OrderEvent::CANCEL_ACCEPTED:
      case OrderEvent::CANCEL_CONFIRMED:
      case OrderEvent::CANCEL_REJECTED:
      case OrderEvent::ORDER_AUTO_CANCELED:
      case OrderEvent::ORDER_FILLED:
        break;
      default:
        return;
    }
    const auto& account_request = order_event_proto.account_request();
    const auto& system_name = account_request.system_name();
    const auto mea = MarketExchangeApi::FromString(fmt::format(
        "{}.{}.{}",
        account_request.market_type(),
        account_request.exchange(),
        account_request.api_version()));
    // 1. update local order context manager first
    auto* ocm = GetOrCreateOrderContextManager(system_name, mea);
    ocm->onOrderEvent(order_event);
    // 2. publish order event
    if (!order_event.has_symbol() or order_event.symbol().empty()) {
      py::print(fmt::format(
          "DEBUG: empty absolute norm for order: \n{}",
          order_event_proto.DebugString()));
    }
    auto timestamp = GetCurrentTimestamp();
    auto* upd_mgr = GetOrCreateOrderUpdateManager(
        system_name,
        mea,
        order_event.symbol() /* absolute_norm */,
        timestamp);
    const auto& native_symbol = order_event.native_symbol();
    if (!upd_mgr->is_symbol_registered(native_symbol)) {
      auto* product = GetOrCreateProduct(mea, order_event.symbol(), timestamp);
      upd_mgr->AddProduct(product);
      upd_mgr->AddNativeSymbol(native_symbol, this);
    }
    using coin2::exchange::base::executor::TopicRecordPacket;
    TopicRecordPacket packet;
    packet.timestamp = timestamp;
    upd_mgr->onBeginPacket(packet);
    upd_mgr->GetProductOrderInfoMutable(native_symbol)->set_event(&order_event);
    upd_mgr->PublishAccountOrder(order_event);
    upd_mgr->onEndPacket(packet);
  }

  auto GetOrCreateRiskUpdateManager(const std::string& system_name, const MarketExchangeApi& mea)
      -> RiskUpdateManager* {
    const auto key = std::pair<std::string, MarketExchangeApi>(system_name, mea);
    auto iter = risk_upd_mgr_map_.find(key);
    if (iter != risk_upd_mgr_map_.end()) {
      return iter->second;
    }
    auto mgr = std::make_unique<RiskUpdateManager>(system_name, mea);
    mgr->RegisterSubsciber(this);
    risk_upd_mgr_map_.emplace(key, mgr.get());
    risk_upd_mgr_holder_.emplace_back(std::move(mgr));
    return risk_upd_mgr_holder_.back().get();
  }

  void HanddleOrderSnapshot(const OrderSnapshotProto& order_snapshot_proto) {
    for (const auto& order_context_bundle : order_snapshot_proto.each_order_context_bundle()) {
      const auto& account_request = order_context_bundle.account_request();
      const auto& system_name = account_request.system_name();
      const auto mea = MarketExchangeApi::FromString(fmt::format(
          "{}.{}.{}",
          account_request.market_type(),
          account_request.exchange(),
          account_request.api_version()));
      auto* ocm = GetOrCreateOrderContextManager(system_name, mea);
      ocm->onOrderConextUpdate(order_context_bundle);
    }
  }

  void HandleOrderUpdate(const OrderUpdateProto& upd) {
    // py::print(fmt::format("DEBUG: order: \n{}", upd.DebugString()));
    switch (upd.type()) {
      case OrderUpdateProto::ORDER_EVENT:
        HandleOrderEvent(upd.order_event());
        break;
      case OrderUpdateProto::ORDER_SNAPSHOT:
        HanddleOrderSnapshot(upd.order_snapshot());
        break;
      default:
        break;
    }
  }

  void HandleRiskUpdate(const RiskUpdateProto& upd) {
    // py::print(fmt::format("@daniel: handle risk update:", upd.DebugString()));
    const auto& account_request = upd.account_request();
    const auto& system_name = account_request.system_name();
    const auto mea = MarketExchangeApi::FromString(fmt::format(
        "{}.{}.{}",
        account_request.market_type(),
        account_request.exchange(),
        account_request.api_version()));
    auto* upd_mgr = GetOrCreateRiskUpdateManager(system_name, mea);
    upd_mgr->HandleRiskUpdate(upd);
  }

  /**************************************************************************************/
  /* interface and strat on callback implementation                                     */
  /**************************************************************************************/
 private:
  // system
  void onPollFail() {
    if (!py::hasattr(strat_, "on_poll_fail")) {
      return;
    }
    strat_.attr("on_poll_fail");  // inject to handle sub-process dead case
  }
  // feed
  void onBookFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_book_feed")) {
      return;
    }
    PyBookView book(&upd);
    strat_.attr("on_book_feed")(book);
  }

  void onTradeFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_trade_feed")) {
      return;
    }
    PyTradeView trade(&upd);
    strat_.attr("on_trade_feed")(trade);
  }
  void onLiquidationFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_liquidation_feed")) {
      return;
    }
    PyLiquidationView liquid(&upd);
    strat_.attr("on_liquidation_feed")(liquid);
  }
  void onOpenInterestFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_open_interest_feed")) {
      return;
    }
    PyOpenInterestView open_interest(&upd);
    strat_.attr("on_open_interest_feed")(open_interest);
  }
  void onMarkPriceFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_mark_price_feed")) {
      return;
    }
    PyMarkPriceView mark_price(&upd);
    strat_.attr("on_mark_price_feed")(mark_price);
  }
  void onIndexFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_index_feed")) {
      return;
    }
    PyIndexView index(&upd);
    strat_.attr("on_index_feed")(index);
  }
  void onFundingRateFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_funding_rate_feed")) {
      return;
    }
    PyFundingRateView funding_rate(&upd);
    strat_.attr("on_funding_rate_feed")(funding_rate);
  }
  void onIvGreeksFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_iv_greeks_feed")) {
      return;
    }
    PyIvGreeksView iv_greeks(&upd);
    strat_.attr("on_iv_greeks_feed")(iv_greeks);
  }
  void onNavFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_nav_feed")) {
      return;
    }
    PyNavView nav(&upd);
    strat_.attr("on_nav_feed")(nav);
  }
  void onKlineFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_kline_feed")) {
      return;
    }
    PyKlinesView kline(&upd);
    strat_.attr("on_kline_feed")(kline);
  }
  void onWalletFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_wallet_feed")) {
      return;
    }
    PyWalletView wallet(&upd);
    strat_.attr("on_wallet_feed")(wallet);
  }
  void onMarketAnalysisFeed(const FeedUpdate& upd) override {
    if (!py::hasattr(strat_, "on_market_analysis_feed")) {
      return;
    }
    PyMarketAnalysisView market_analysis(&upd);
    strat_.attr("on_market_analysis_feed")(market_analysis);
  }
  void onHeartbeatFeed(const FeedTimer& timer) override {}
  // order
  void onAccountOrder(const OrderUpdate& upd) override {
    if (!py::hasattr(strat_, "on_account_order")) {
      return;
    }
    PyOrderUpdate order(&upd);
    strat_.attr("on_account_order")(order);
  }

  void onAccountInfo(const OrderUpdate& upd) override {
    if (!py::hasattr(strat_, "on_account_info")) {
      return;
    }
    PyAccountInfo info(&upd);
    strat_.attr("on_account_info")(info);
  }

  void onOrderLog(const OrderGatewayLog& log) override {
    if (log.type() == OrderGatewayLog::RISK_EVENT) {
      if (py::hasattr(strat_, "on_account_risk")) {
        PyAccountRisk info(&log);
        strat_.attr("on_account_risk")(info);
      }
    }
  }

 private:
  py::object strat_;
  py::object pipe_incoming_;
  py::object pipe_outgoing_;
  bool running_ = true;

  // message
  std::string send_buf_;
  std::string recv_buf_;
  CoinRequestProto req_proto_;
  CoinUpdateProto upd_proto_;

  // TODO(daniel): feed update manager
  std::vector<std::unique_ptr<FeedUpdateManager>> feed_upd_mgr_holder_;
  std::map<UpdateManagerKey, FeedUpdateManager*> feed_upd_mgr_map_;

  // TODO(daniel): order manager
  // - order update manager
  std::vector<std::unique_ptr<OrderUpdateManager>> order_upd_mgr_holder_;
  std::map<UpdateManagerKey, OrderUpdateManager*> order_upd_mgr_map_;
  // - risk update manager
  std::vector<std::unique_ptr<RiskUpdateManager>> risk_upd_mgr_holder_;
  std::map<UpdateManagerKey, RiskUpdateManager*> risk_upd_mgr_map_;
  // - order context manager
  std::vector<std::unique_ptr<OrderContextManager2>> order_context_manager_holder_;
  std::map<std::string, OrderContextManager2*> order_context_manager_map_;
  bool is_order_updated_ = true;
  std::vector<PyOrderContext> working_order_;

  // TODO(daniel): product
  std::vector<std::unique_ptr<IProduct>> local_product_holder_;
  std::map<std::pair<std::string, MarketExchangeApi>, IProduct*> symbol_to_product_map_;
};

inline void AddPyStratForwarder(py::module* m) {
  py::class_<PyStratForwarder, std::shared_ptr<PyStratForwarder>>(*m, "PyStratForwarder")
      .def(py::init<py::object /* strat */, py::object /* pipe */, py::object /* pipe */>())
      .def("run", &PyStratForwarder::Run)
      .def("stop", &PyStratForwarder::Stop)
      .def("get_all_working_orders", &PyStratForwarder::GetAllWorkingOrders)
      .def("submit_execution", &PyStratForwarder::SubmitExecution);
}

}  // namespace impl

using impl::PyStratForwarder;
}  // namespace coin2::exchange::driver
